package com.atguigu.gmall.realtime.utils;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;

/**
 * @author Felix
 * @date 2023/12/16
 * 操作kafka的工具类
 * 在生产环境，要向保证发送到kafka的数据的精准一次性，需要做如下的设置
 * 开启检查点
 * 设置开启精准一次，底层开启2pc提交事务
 * .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
 * 设置事务id前缀
 * .setTransactionalIdPrefix("dwd_traffic_dirty_data")
 * 设置事务超时时间    检查点超时时间 <   transaction.timeout.ms  <=transaction.max.timeout.ms
 * .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 + "")
 * 在消费端，预提交的数据不能被消费，需要设置事务的隔离级别
 * .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")
 */
public class MyKafkaUtil {
    private static final String KAFKA_SERVER = "hadoop102:9092,hadoop103:9092,hadoop104:9092";

    //获取KafkaSource
    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
            .setBootstrapServers(KAFKA_SERVER)
            .setTopics(topic)
            .setGroupId(groupId)
            //在生产环境中，要向保证消费的精准一次，需要做如下设置，表示先从维护的偏移量开始消费，如果没找到，再从kafka最新偏移量消费
            // .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
            //在学习的时候，我们直接从kafka的最新偏移量开始读取
            .setStartingOffsets(OffsetsInitializer.latest())
            //在生产环境中，要想保证生产端的精准一次，消费端不能消费预提交数据，需要设置隔离级别为读已提交
            // .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")
            //默认使用SimpleStringSchema，对kafka消息进行反序列化的时候，如果消息为空，那么处理不了，我们需要自定义反序列化器
            // .setValueOnlyDeserializer(new SimpleStringSchema())
            .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                @Override
                public String deserialize(byte[] message) throws IOException {
                    if (message != null) {
                        return new String(message);
                    }
                    return null;
                }

                @Override
                public boolean isEndOfStream(String nextElement) {
                    return false;
                }

                @Override
                public TypeInformation<String> getProducedType() {
                    return TypeInformation.of(String.class);
                }
            })
            .build();
        return kafkaSource;
    }

    //获取KafkaSink
    public static KafkaSink<String> getKafkaSink(String topic) {
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
            .setBootstrapServers(KAFKA_SERVER)
            .setRecordSerializer(
                KafkaRecordSerializationSchema.<String>builder()
                    .setTopic(topic)
                    .setValueSerializationSchema(new SimpleStringSchema())
                    .build()
            )
            // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            // .setTransactionalIdPrefix("dwd_traffic_dirty_data")
            // .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 + "")
            .build();
        return kafkaSink;
    }

    //获取KafkaSink,可以将流的数据发送到kafka的不同的主题
    public static <T>KafkaSink<T> getKafkaSinkBySchema(KafkaRecordSerializationSchema<T> krs) {
        KafkaSink<T> kafkaSink = KafkaSink.<T>builder()
            .setBootstrapServers(KAFKA_SERVER)
            .setRecordSerializer(krs)
            // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            // .setTransactionalIdPrefix("dwd_traffic_dirty_data")
            // .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 + "")
            .build();
        return kafkaSink;
    }

    //获取从topic_db主题中读取数据 创建动态表的建表语句
    public static String getTopicDbDDL(String groupId) {
        return "CREATE TABLE topic_db (\n" +
            "  `database` string,\n" +
            "  `table` string,\n" +
            "  `type` string,\n" +
            "  `ts`  string,\n" +
            "  `data` MAP<string, string>,\n" +
            "  `old` MAP<string, string>,\n" +
            "  proc_time as proctime()\n" +
            ") " + getKafkaDDL("topic_db", groupId);
    }

    //获取kafka连接器的连接属性
    public static String getKafkaDDL(String topic, String groupId) {
        return " WITH (\n" +
            "  'connector' = 'kafka',\n" +
            "  'topic' = '" + topic + "',\n" +
            "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
            "  'properties.group.id' = '" + groupId + "',\n" +
            //"  在生产环境，消费起始位点设置为group-offsets；如果没有找到可以设置kafka的auto.offset.reset\n" +
            //"  'scan.startup.mode' = 'group-offsets',\n" +
            //"  'properties.auto.offset.reset' = 'latest'\n" +
            //"  在学习阶段，消费起始位点设置为latest-offset\n" +
            "  'scan.startup.mode' = 'latest-offset',\n" +
            "  'format' = 'json'\n" +
            ")";
    }

    //获取upsert-kafka连接器的连接属性
    public static String getUpsertKafkaDDL(String topic) {
        return " WITH (\n" +
            "  'connector' = 'upsert-kafka',\n" +
            "  'topic' = '" + topic + "',\n" +
            "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
            "  'key.format' = 'json',\n" +
            "  'value.format' = 'json'\n" +
            ")";
    }
}
